package jmine.tec.hydra.receive.impl.mq;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Map;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

import jmine.tec.hydra.HydraController;
import jmine.tec.hydra.HydraMessages;
import jmine.tec.hydra.domain.HydraSystem;
import jmine.tec.hydra.exception.HydraException;
import jmine.tec.hydra.util.stream.InputableOutputStream;

import org.apache.log4j.Logger;

import bancosys.tec.persist.dao.BaseDAO;
import bancosys.tec.persist.dao.BeanNotFoundException;
import bancosys.tec.rtm.impl.RtmController;

/**
 * @author takeshi
 */
public class HydraMessageListener implements MessageListener {

    private static final int BUFFER_SIZE = 4048;

    public static final String MESSAGE_PUT_TIMESTAMP = "MESSAGE_PUT_TIMESTAMP";

    private static final Logger LOGGER = Logger.getLogger(HydraMessageListener.class);

    private HydraController hydraController;

    private RtmController rtmController;

    private String systemName;

    private HydraSystem system;

    /**
     * Chamado pelo container para tratar as mensagens de JMS.
     * 
     * @param msg mensagem jms
     */
    public void onMessage(Message msg) {
        String messageId = this.getMessageId(msg);
        LOGGER.debug("Mensagem recebida pelo Hydra. MessageId:" + messageId);

        Map<String, String> properties = null;
        String ts = this.getPutTimestamp(msg);
        if (ts != null) {
            properties = new HashMap<String, String>();
            properties.put(MESSAGE_PUT_TIMESTAMP, ts);
        }

        boolean commit = false;
        try {
            this.hydraController.getPersistenceController().startTransaction();

            // Lê a mensagem
            InputStream input;
            if (msg instanceof BytesMessage) {
                input = this.readMessage((BytesMessage) msg, messageId);
            } else {
                input = new ByteArrayInputStream(((TextMessage) msg).getText().getBytes());
            }

            this.hydraController.receive(this.getSystem(), input, properties);

            commit = true;
        } catch (RuntimeException e) {
            this.rtmController.monitor(e);
            throw e;
        } catch (Exception e) {
            this.rtmController.monitor(e);
            throw new RuntimeException(e);
        } finally {
            if (commit) {
                this.hydraController.getPersistenceController().commit();
            } else {
                this.hydraController.getPersistenceController().rollback();
            }
        }

    }

    /**
     * Devolve o MessageId de uma mensagem.
     * 
     * @param message mensagem
     * @return messageId
     */
    private String getMessageId(Message message) {
        try {
            return message.getJMSMessageID();
        } catch (JMSException e) {
            throw new RuntimeException("Erro ao obter MessageId.", e);
        }
    }

    /**
     * Retorna a data/hora de postagem da mensagem.
     * 
     * @param message mensagem
     * @return data/hora de postagem da mensagem.
     */
    private String getPutTimestamp(Message message) {
        final SimpleDateFormat sdf = new SimpleDateFormat("dd/MM/yyyy hh:mm:ss Z");
        try {
            long timestamp = message.getJMSTimestamp();
            return sdf.format(new java.util.Date(timestamp));
        } catch (JMSException e) {
            LOGGER.warn("Nao foi possivel obter o timestamp de postagem (JMSTimestamp/PutTimestamp) da mensagem.", e);
            return null;
        }
    }

    /**
     * Le os dados da mensagem.
     * 
     * @param message mensagem
     * @param messageId id da mensagem
     * @return os dados da mensagem
     * @throws IOException
     */
    private InputStream readMessage(BytesMessage message, String messageId) {
        InputableOutputStream os = new InputableOutputStream();
        byte[] buffer = new byte[BUFFER_SIZE];
        int bytesRead = 0;
        try {
            while (bytesRead != -1) {
                try {
                    bytesRead = message.readBytes(buffer);
                } catch (JMSException e) {
                    throw new RuntimeException("Erro ao ler arquivo da mensagem. MessageId: " + messageId, e);
                }
                if (bytesRead != -1) {
                    os.write(buffer, 0, bytesRead);
                }
            }
            return os.getAsInputStream();
        } catch (IOException e) {
            throw new RuntimeException("Erro ao converter mensagem. MessageId: " + messageId, e);
        }
    }

    /**
     * @return o valor do campo system
     */
    private HydraSystem getSystem() {
        if (this.system == null) {
            try {
                BaseDAO<HydraSystem> sysDAO = this.hydraController.getDAOFactory().getGenericDAO(HydraSystem.class);
                this.system = sysDAO.findByNaturalKey(this.systemName);
            } catch (BeanNotFoundException e) {
                throw new HydraException(HydraMessages.RECEIVE_MQ_RECEIVER.create(this.systemName), e);
            }
        }
        return this.system;
    }

    /**
     * @param hydraController o valor para o campo hydraController
     */
    public void setHydraController(HydraController hydraController) {
        this.hydraController = hydraController;
    }

    /**
     * @param rtmController o valor para o campo rtmController
     */
    public void setRtmController(RtmController rtmController) {
        this.rtmController = rtmController;
    }

    /**
     * @param systemName o valor para o campo systemName
     */
    public void setSystemName(String systemName) {
        this.systemName = systemName;
    }

    /**
     * @param system the system to set
     */
    public void setSystem(HydraSystem system) {
        this.system = system;
    }

}
